package com.cssl.app.function;

import com.alibaba.fastjson.JSONObject;
import com.cssl.utils.CommonUtils;
import com.cssl.utils.DimUtil;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.Collection;
import java.util.Properties;
import java.util.Set;

/**
 * @Author: chen
 * @Date: 2021/11/12 23:33
 * @Desc: 将 dim 表 upsert 到 hbase
 */
public class DimSinkFunction extends RichSinkFunction<JSONObject> {
    private Properties properties;
    private Connection connection;

    @Override
    public void open(Configuration parameters) throws Exception {
        properties = CommonUtils.getProperties();
        String driver = properties.getProperty("phoenix.driver");
        String url = properties.getProperty("phoenix.url");
        Class.forName(driver);
        connection = DriverManager.getConnection(url);
        connection.setAutoCommit(true);

    }

    @Override
    public void invoke(JSONObject value, Context context) throws Exception {
        //获取sql
        PreparedStatement preparedStatement = null;
        try {
            String sinkTable = value.getString("sinkTable");
            JSONObject after = value.getJSONObject("after");
            String upsertSql = getUpsertSql(
                    sinkTable,
                    after
            );
            System.out.println("执行插入语句: " + upsertSql);

            //预编译sql
            preparedStatement = connection.prepareStatement(upsertSql);

            //判断如果当前数据为更新操作,则先删除redis中的数据
            if ("update".equals(value.getString("type"))) {
                DimUtil.delRedisDimInfo(sinkTable, after.getString("id"));
            }

            //执行插入操作
            preparedStatement.executeUpdate();
        } catch (SQLException throwables) {
            throwables.printStackTrace();
        } finally {
            if (preparedStatement != null) {
                preparedStatement.close();
            }
        }
    }

    private String getUpsertSql(String sinkTable, JSONObject after) {
        String database = properties.getProperty("phoenix.database");
        Set<String> keySet = after.keySet();
        Collection<Object> values = after.values();

        return "upsert into " + database + "." + sinkTable + "(" +
                StringUtils.join(keySet, ",") + ")values('" +
                StringUtils.join(values, "','") + "')";
    }
}
